package com.paratice.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class DistributeLock implements Lock, Watcher {
	
	private ZooKeeper zk = null;
	private String ROOT_LOCK = "/locks";
	private String WAIT_LOCK;
	private String CURRENT_LOCK;
	
	private CountDownLatch countDownLatch;

	public DistributeLock() {
		try {
			zk = new ZooKeeper("192.168.2.200:2181", 4000, this);
			Stat stat = zk.exists(ROOT_LOCK, false);
			if (stat == null) {
				zk.create(ROOT_LOCK, "0".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		} catch(Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void process(WatchedEvent event) {
		if (this.countDownLatch != null) {
			countDownLatch.countDown();
		}
	}

	@Override
	public void lock() {
		if (this.tryLock()) {
			System.out.println(Thread.currentThread().getName() + "->" + CURRENT_LOCK + "获得锁成功");
			return;
		}
		waitForLock(WAIT_LOCK);
	}
	
	private void waitForLock(String prev) {
		try {
			Stat stat = zk.exists(prev, true);
			if (stat != null) {
				System.out.println(Thread.currentThread().getName() + "->等待锁" + prev + "释放");
				countDownLatch = new CountDownLatch(1);
				countDownLatch.await();
				System.out.println(Thread.currentThread().getName() + "获得锁成功");
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean tryLock() {
		try {
			CURRENT_LOCK = zk.create(ROOT_LOCK + "/", "0".getBytes(), 
					Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
			System.out.println(Thread.currentThread().getName() + "->" + CURRENT_LOCK + ", 尝试竞争锁");
			
			List<String> childrens = zk.getChildren(ROOT_LOCK, false);
			SortedSet<String> sortedSet = new TreeSet<>();
			for (String child : childrens) {
				sortedSet.add(ROOT_LOCK + "/" + child);
			}
			
			String firstNode = sortedSet.first();
			SortedSet<String> lessThanMe = sortedSet.headSet(CURRENT_LOCK);
			if (CURRENT_LOCK.equals(firstNode)) {
				return true;
			}
			
			if (!lessThanMe.isEmpty()) {
				WAIT_LOCK = lessThanMe.last();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}
	
	@Override
	public void unlock() {
		System.out.println(Thread.currentThread().getName() + "->释放锁" + CURRENT_LOCK);
		try {
			zk.delete(CURRENT_LOCK, -1);
			CURRENT_LOCK = null;
			zk.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}
	

	@Override
	public Condition newCondition() {
		return null;
	}
	
	@Override
	public void lockInterruptibly() throws InterruptedException {
		
	}
	
	public static void main(String[] args) {
		CountDownLatch countDownLatch = new CountDownLatch(10);
		for (int i = 0; i < 10; i++) {
			new Thread(()->{
				try {
					countDownLatch.await();	
					DistributeLock lock = new DistributeLock();
					lock.lock();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}, "Thread" + i).start();
			countDownLatch.countDown();
		}
		try {
			System.in.read();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
